package com.develop.collection.middle.sensors;

import com.develop.collection.util.Entry;
import com.develop.jdbc.jdbc.DataRow;
import com.develop.jdbc.jdbc.session.Session;
import com.develop.jdbc.jdbc.session.SessionFactory;
import com.develop.jdbc.util.DateHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author fenny
 * @since 2019/08/19
 */
public class SensorsData {

    private static final Logger log = LoggerFactory.getLogger(SensorsData.class);

    private static final String TABLE_NAME = "kilimall_events";

    private static final String USER_TABLE_NAME = "kilimall_user";

    private static final AtomicInteger SUCCESS_COUNT = new AtomicInteger();

    private static final AtomicInteger FILTER_COUNT = new AtomicInteger();

    private static final AtomicInteger CONVERT_COUNT = new AtomicInteger();

    private static final SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DateHelper.pattern_date);

    private static final String PROJECT = "bbc_ke";

    private static void execute() {
        log.info("begin-采集");
        String currDate = "2018-06-15";
        currDate = "2018-07-03";
        /*currDate = "2018-09-23";
        currDate = "2018-10-27";
        currDate = "2018-11-14";
        currDate = "2019-02-19";
        currDate = "2019-03-11";
        currDate = "2019-04-08";
        currDate = "2019-05-31";
        currDate = "2019-08-22";
        currDate = "2019-08-29";
        currDate = "2019-08-30";
        currDate = "2019-09-03";
        currDate = "2019-09-04";
        */
        currDate = "2019-09-06";
        currDate = "2018-11-13";
        ExecutorService pagePool = Executors.newFixedThreadPool(10);
        ExecutorService pool = Executors.newFixedThreadPool(1000);
        int currTotal = 0;
        int pageSize = 10000;
        int totalPage;
        long beginTime;
        try {
            int sumDay = DateHelper.getDateDiff(new Date(), simpleDateFormat.parse(currDate));
//            sumDay = 1;
            log.info("总天数={}", sumDay);
            //第几天
            for (int i = 1; i <= sumDay; i++) {
                beginTime = System.currentTimeMillis();
                Session source = SessionFactory.getSession("sensors");
                //获得当天总记录数
                currTotal = source.queryInt(String.format("select count(1) c from (select a.event,b.second_id, a.commodityid from events a inner join users b on(a.user_id=b.id) where a.date = '%s' and a.commodityid is not null and b.second_id is not null group by a.event,b.second_id, a.commodityid) a /*SA(%s)*/;", currDate, PROJECT));
                source.close();
                totalPage = (currTotal + pageSize - 1) / pageSize;
                log.info("{},共{}条,共{}页,每页{}条", currDate, currTotal, totalPage, pageSize);
                //第几页
                CountDownLatch pageLatch = new CountDownLatch(totalPage);
                for (int j = 1; j <= totalPage; j++) {
                    //开始分页处理
                    int finalJ = j;
                    String finalCurrDate = currDate;
                    pagePool.submit(() -> {
                        try {
                            int offset = (finalJ - 1) * pageSize;
                            Session pageSource = SessionFactory.getSession("sensors");
                            List<DataRow> currResult = pageSource.query(String.format("select a.event,b.second_id, a.commodityid,count(1) c from events a inner join users b on(a.user_id=b.id) where a.date = '%s' and a.commodityid is not null and b.second_id is not null group by a.event,b.second_id, a.commodityid order by a.event,b.second_id, a.commodityid limit %s,%s /*SA(%s)*/;", finalCurrDate, offset, pageSize, PROJECT));
                            pageSource.close();
                            CountDownLatch latch = new CountDownLatch(currResult.size());
                            for (DataRow r : currResult) {
                                pool.submit(() -> {
                                    Session target = null;
                                    DataRow row = new DataRow();
                                    try {
                                        target = SessionFactory.getSession("remote-mysql");
                                        String event = r.getString("event");
                                        long userId = r.getLong("second_id");
                                        long skuId = r.getLong("commodityid");
                                        int num = r.getInt("c");
                                        if (userId > 0 && skuId > 0) {
                                            int count = target.queryInt(String.format("select count(1) c from %s where event='%s' and user_id=%s and sku_id=%s", TABLE_NAME, event, userId, skuId));
                                            if (count == 1) {
                                                row.set("event", event);
                                                row.set("user_id", userId);
                                                row.set("sku_id", skuId);
                                                row.set("num", r.getInt("c"));
                                                target.update(String.format("update %s set num=num+%s where event='%s' and user_id=%s and sku_id=%s", TABLE_NAME, num, event, userId, skuId));
                                            } else {
                                                row.set("event", event);
                                                row.set("user_id", userId);
                                                row.set("sku_id", skuId);
                                                row.set("num", r.getInt("c"));
                                                target.insert(TABLE_NAME, row);
                                            }
                                            int userCount = target.queryInt(String.format("select count(1) c from %s where user_id=%s", USER_TABLE_NAME, userId));
                                            if (userCount == 0) {
                                                try {
                                                    DataRow userRow = new DataRow();
                                                    userRow.set("user_id", userId);
                                                    target.insert(USER_TABLE_NAME, userRow);
                                                } catch (Exception ignored) {
                                                }
                                            }
                                            SUCCESS_COUNT.incrementAndGet();
                                        } else {
                                            //用户ID与产品ID不是数字过滤掉
                                            FILTER_COUNT.incrementAndGet();
                                        }
                                    } catch (Exception e) {
                                        log.error("{},第{}页,{}条。异常数据={}", finalCurrDate, finalJ, currResult.size(), row);
                                        log.error("", e);
                                    } finally {
                                        latch.countDown();
                                        if (target != null) {
                                            target.close();
                                        }
                                    }
                                });
                            }
                            try {
                                latch.await();
                            } catch (InterruptedException e) {
                                log.error("", e);
                            }
                            log.info("完成第{}页", finalJ);
                        } catch (Exception e) {
                            log.error("", e);
                        } finally {
                            pageLatch.countDown();
                        }
                    });
                }
                try {
                    pageLatch.await();
                } catch (InterruptedException e) {
                    log.error("", e);
                } finally {
                    log.info("耗时{}_{}秒", currDate, ((System.currentTimeMillis() - beginTime) / 1000));
                    currDate = getDate(currDate, 1);
                }
            }
        } catch (Exception e) {
            log.error("", e);
        } finally {
            pool.shutdown();
            pagePool.shutdown();
            log.info("原数据量={},过滤量={},写入量={},转换量={}", currTotal, FILTER_COUNT.get(), SUCCESS_COUNT.get(), CONVERT_COUNT.get());
            log.info("end-采集");
        }
    }

    public static String getDate(String date, int day) throws ParseException {
        Date currdate = simpleDateFormat.parse(date);
        Calendar ca = Calendar.getInstance();
        ca.setTime(currdate);
        ca.add(Calendar.DATE, day);
        return simpleDateFormat.format(ca.getTime());
    }

    public static void main(String[] args) {
        Entry.entry(SensorsData::execute);
    }


}
